using System;
using System.Collections.Generic;
using System.ComponentModel.Composition;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.Collections.Concurrent;

namespace FinMkts.Messaging.Pgm
{
    /// <summary>
    /// Publishes <see cref="AddressedMessage"/> using the PGM protocol (install MSMQ multicast in Windows setup)
    /// </summary>
    /// <remarks>
    /// Publishing appears to crash Windows if more than one thread publishes on a socket,
    /// therefore all work is queued and sent using a single thread
    /// </remarks>
    public class PgmPublisher : Publisher
    {
        private Socket _socket;
        readonly CancellationTokenSource _cancelSource = new CancellationTokenSource();
        readonly Subject<RmSenderStats> _statsSubject = new Subject<RmSenderStats>();
        private Timer _timer;
        private SocketAsyncEventArgs _asyncEventArgs;
        private BlockBufferedBlockingQueue _queue;

        [Import]
        public PgmConfiguration Config { get; set; }

        [Import]
        public AddressedMessageSerializer Serializer { get; set; }

        public IObservable<RmSenderStats> SenderStatistics
        {
            get { return _statsSubject; }
        }

        public Socket Socket 
        { 
            get { return _socket; } 
        }

        public override void Start()
        {
            if (Config == null) throw new InvalidOperationException("Config must be set before Start is called");
            if (Serializer == null) throw new InvalidOperationException("Serializer must be set before Start is called");

            // create socket and allow others to publish on the same multicast group
            _socket = new Socket(AddressFamily.InterNetwork, SocketType.Rdm, RmConstants.ProtocolType);
            _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
            _socket.Bind(new IPEndPoint(IPAddress.Any, 0));

            // set options before connect
            _socket.SetSocketOption(RmSocketOptionName.SetTimeToLive, Config.MulticastTTL);
            var sendWindow = new RmSendWindow
                                   {
                                       RatePerSec = Config.RatePerSec,
                                       WindowSize = Config.WindowSize,
                                       WindowLength = Config.WindowLength,
                                   };
            Debug.WriteLine(sendWindow);
            _socket.SetSendWindow(sendWindow);
            if (!string.IsNullOrEmpty(Config.BindInterface))
            {
                var ip = IPAddress.Parse(Config.BindInterface);
                _socket.SetSocketOption(RmSocketOptionName.SetSendInterface, ip);
            }
            
            Connect();

            _asyncEventArgs.Completed += SendComplete;
            _queue = new BlockBufferedBlockingQueue(Config.MaxMessageSize, BlockReadyToSend);

            if (Config.StatsInterval > TimeSpan.Zero)
            {
                _timer = new Timer(PublishStatistics, null, Config.StatsInterval, Config.StatsInterval);
            }
        }

        public override void Stop()
        {
            if (_socket == null) return;
            if (_timer != null)
            {
                _timer.Dispose();
            }

            _asyncEventArgs.Completed -= SendComplete;
            _queue = null;

            _cancelSource.Cancel();
            _socket.Shutdown(SocketShutdown.Send);
            _socket.Close();
            _statsSubject.OnCompleted();            
        }

        public override void Publish(IEnumerable<AddressedMessage> messages)
        {
            foreach (var message in messages)
            {
                Publish(message);
            }
        }

        public override void Publish(AddressedMessage message)
        {
            var serialized = Serializer.Serialize(message);
            _queue.Add(serialized);
        }

        private void Connect()
        {
            using (var connected = new ManualResetEvent(false))
            {
                _asyncEventArgs = new SocketAsyncEventArgs { RemoteEndPoint = new IPEndPoint(Config.MulticastAddress, Config.MulticastPort) };
                EventHandler<SocketAsyncEventArgs> completed = (sender, args) => connected.Set();
                _asyncEventArgs.Completed += completed;
                var pending = _socket.ConnectAsync(_asyncEventArgs);
                if (pending)
                {
                    connected.WaitOne();
                }
                _asyncEventArgs.Completed -= completed;
            }
        }

        private void BlockReadyToSend(ByteBufferCollection buffers)
        {
            lock (_socket)
            {
                while (buffers != null)
                {
                    _asyncEventArgs.BufferList = buffers;
                    bool pending = _socket.SendAsync(_asyncEventArgs);
                    _queue.IsBlocked = pending;
                    if (pending)
                    {
                        return;
                    }
                    if (_asyncEventArgs.SocketError != SocketError.Success)
                    {
                        //TODO: error! do not throw expection here!!!!
                        throw new SocketException((int)_asyncEventArgs.SocketError);
                    }
                    buffers = _queue.TakeNextBlock();
                }

                // nothing to send and no send is pending
                _queue.IsBlocked = false;
            }
        }

        private void SendComplete(object sender, SocketAsyncEventArgs e)
        {
            if (e.SocketError != SocketError.Success)
            {
                //TODO: error! do not throw expection here!!!!
                throw new SocketException((int)_asyncEventArgs.SocketError);
            }

            BlockReadyToSend(_queue.TakeNextBlock());
        }

        private void PublishStatistics(object state)
        {
            RmSenderStats stats;
            try
            {
                stats = _socket.GetSenderStats();
            }
            catch (Exception e)
            {
                _statsSubject.OnError(e);
                return;
            }

            try
            {
                _statsSubject.OnNext(stats);
            }
            catch
            {
                // ignore errors
            }
        }
    }
}